package org.example.utils;

import org.apache.spark.sql.*;
import org.apache.spark.sql.types.StructField;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.sql.*;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * MysqlUtils_2 工具类，用于读取 MySQL 数据表
 */
public class MysqlUtils_2 {
    private static ConfigProperties configProperties;
    private static String url;
    private static String user;
    private static String password;
    private static String driver;

    static {
        configProperties = ConfigProperties.getInstance();
        url = configProperties.getProperty("jmis_mysql_url");
        user = configProperties.getProperty("jmis_mysql_user");
        password = configProperties.getProperty("jmis_mysql_pwd");
        driver = configProperties.getProperty("mysql_driver");
    }

    /**
     * 初始化 MySQL 连接并将数据表作为临时视图加载到 Spark 中
     *
     * @param spark          SparkSession 对象
     * @param mysqlTableName MySQL 数据表名称
     * @return String 临时视图名称
     * @throws IOException 抛出文件未找到异常
     */
    public static String readMysqlToTempTable(SparkSession spark, String mysqlTableName) throws IOException {
        String tempTableName = mysqlTableName;
        spark.read().format("jdbc")
                .option("url", url)
                .option("user", user)
                .option("password", password)
                .option("dbtable", mysqlTableName)
                .option("dateFormat", "yyyy-MM-dd") // 根据需要设置日期格式
                .option("timestampFormat", "yyyy-MM-dd HH:mm:ss") // 根据需要设置时间戳格式
                .load().createOrReplaceTempView(tempTableName);
        return tempTableName;
    }

    /**
     * 读取 MySQL 数据表
     *
     * @param spark     SparkSession 对象
     * @param tableName MySQL 数据表名称
     * @return Dataset<Row> 返回数据集
     */
    public static Dataset<Row> readMysqlTableDs(SparkSession spark, String tableName) throws IOException {
        Dataset<Row> dataSet = spark.read().format("jdbc")
                .option("url", url)
                .option("driver", driver)
                .option("dbtable", tableName)
                .option("user", user)
                .option("password", password)
                .option("dateFormat", "yyyy-MM-dd") // 根据需要设置日期格式
                .option("timestampFormat", "yyyy-MM-dd HH:mm:ss") // 根据需要设置时间戳格式
                .load();
        return dataSet;
    }

    /***
     * 定义一个工具类,读取MySQL的指定表名和JavaBean,返回一个DataSet<Bean>类型的DF.
     * @param spark
     * @param tableName 表名称
     * @param beanClass 对应表的表结构
     * @return DataSet<Bean>类型的DF
     * @param <T>
     */
    public static <T> Dataset<T> readMySQLTableDF(SparkSession spark, String tableName, Class<T> beanClass) {
        // 从MySQL读取数据
        Dataset<Row> df = spark.read()
                .format("jdbc")
                .option("url", url)
                .option("dbtable", tableName)
                .option("user", user)
                .option("password", password)
                .option("dateFormat", "yyyy-MM-dd") // 根据需要设置日期格式
                .option("timestampFormat", "yyyy-MM-dd HH:mm:ss") // 根据需要设置时间戳格式
                .load();
        // 将DataFrame转换为Dataset<JavaBean>
        return df.as(Encoders.bean(beanClass));
    }

    /**
     * 加载配置文件并返回 Properties 对象
     *
     * @param fileName 配置文件名称
     * @return Properties 对象
     */
    private static Properties loadPropertiesFile(String fileName) throws IOException {
        Properties properties = new Properties();
        InputStream inputStream = MysqlUtils_2.class.getClassLoader().getResourceAsStream(fileName);
        if (inputStream != null) {
            properties.load(inputStream);
        } else {
            throw new FileNotFoundException(fileName + " not found in classpath");
        }
        return properties;
    }

    /**
     * 将数据集写入 MySQL 数据表
     *
     * @param dataSet   要写入的数据集
     * @param tableName MySQL 数据表名称
     * @param saveMode  写入模式（追加、覆盖、忽略、出错）
     *                  <p>
     *                  示例:
     *                  Dataset<Row> dataSet = ...; // 获取数据集
     *                  String tableName = "your_table_name"; // 目标 MySQL 数据表名称
     *                  SaveMode saveMode = SaveMode.Append; // 设置写入模式
     *                  MysqlUtils_2.writeDataToMysqlTable(dataSet, tableName, saveMode); // 写入数据到 MySQL 数据表
     *                  <p>
     *                  写入模式说明:
     *                  - SaveMode.Append: 将数据追加到已存在的表中
     *                  - SaveMode.Overwrite: 覆盖已存在的表中的数据
     *                  - SaveMode.Ignore: 如果表已存在，则忽略写入操作
     *                  - SaveMode.ErrorIfExists: 如果表已存在，则抛出异常
     */
    public static void writeDataToMysqlTable(Dataset<Row> dataSet, String tableName, SaveMode saveMode) throws IOException {
        Properties properties = loadPropertiesFile("config.properties");
        dataSet.write().mode(saveMode)
                .option("url", url)
                .option("driver", driver)
                .option("dbtable", tableName)
                .option("user", user)
                .option("password", password)
                .jdbc(url, tableName, properties);
    }

    /**
     * 更新 MySQL 数据表中的数据
     *
     * @param tableName MySQL 数据表名称
     * @param updateSql 更新数据的 SQL 语句
     * @throws SQLException 抛出 SQL 异常
     *                      <p>
     *                      示例:
     *                      String tableName = "your_table_name"; // 目标 MySQL 数据表名称
     *                      String updateSql = "UPDATE " + tableName + " SET column1 = 'new_value1', column2 = 'new_value2' WHERE id = 1";
     *                      MysqlUtils_2.updateMysqlTableData(tableName, updateSql); // 更新 MySQL 数据表中的数据
     */
    public static void updateMysqlTableData(String tableName, String updateSql) throws SQLException, IOException {
        try (Connection connection = DriverManager.getConnection(url, user, password);
             Statement statement = connection.createStatement()) {
            Class.forName(driver);
            statement.executeUpdate(updateSql);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除 MySQL 数据表中的数据
     *
     * @param tableName MySQL 数据表名称
     * @param deleteSql 删除数据的 SQL 语句
     * @throws SQLException 抛出 SQL 异常
     */
    public static void deleteMysqlTableData(String tableName, String deleteSql) throws SQLException, IOException {

        try (Connection connection = DriverManager.getConnection(url, user, password);
             Statement statement = connection.createStatement()) {
            Class.forName(driver);
            statement.executeUpdate(deleteSql);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    /***
     * 定义一个根据主键更新MySql中表的方法,更新方式: 如果主键存在则更新,如果主键不存在则插入.动态拼接SQL语句.
     * 使用案例:
     * @param dataset
     * @param tableName
     */
    public static void upsertDatasetToMySQL(Dataset<Row> dataset, String tableName) throws IOException {
        StructField[] fields = dataset.schema().fields();
        String[] columnNames = Arrays.stream(fields).map(StructField::name).toArray(String[]::new);
        dataset.foreachPartition(partition -> {
            try (Connection connection = DriverManager.getConnection(url, user, password)) {
                // 遍历每个数据分区
                while (partition.hasNext()) {
                    Row row = partition.next();
                    // 构建 SQL 语句
                    String columns = String.join(", ", columnNames);
                    String placeholders = String.join(", ", Collections.nCopies(columnNames.length, "?"));
                    String updateClause = Arrays.stream(columnNames)
                            .map(column -> column + " = VALUES(" + column + ")")
                            .collect(Collectors.joining(", "));
                    String sql = "INSERT INTO " + tableName + " (" + columns + ") VALUES (" + placeholders + ") " +
                            "ON DUPLICATE KEY UPDATE " + updateClause;
                    try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
                        for (int i = 0; i < columnNames.length; i++) {
                            Object value = row.getAs(columnNames[i]);
                            preparedStatement.setObject(i + 1, value);
                        }
                        preparedStatement.executeUpdate();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}